//
//  ConnectableObservableViewController.swift
//  Example
//
//  Created by T AO on 2022/2/10.
//

import UIKit
import RxSwift

class ConnectableObservableViewController: BaseViewController {

    override func viewDidLoad() {
        super.viewDidLoad()

//        test()
        
//        publish()
//        replay()
        
//        multicast()
//        refCount()
        
        share()
    }

}

extension ConnectableObservableViewController {
    /// 先看下普通序列的
    /// 每隔1s发送1个事件
    private func test() {
        let interval = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance)
        
        // 第一个订阅者（立刻开始订阅）
        _ = interval
            .subscribe(onNext: { print("订阅者1: \($0)") })
        
        // 第二个订阅者 (延迟5s开始订阅)
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅者2: \($0)") })
        }
    }
    
    /// 延迟执行
    /// - Parameters:
    ///   - delay: 延迟时间（秒）
    ///   - closure: 延迟执行闭包
    private func delay(_ delay: Double, closure: @escaping () -> ()) {
        DispatchQueue.main.asyncAfter(deadline: .now() + delay) {
            closure()
        }
    }
}

// MARK: - 连接操作符

extension ConnectableObservableViewController {
    /// publish
    /// 将一个正常的序列转换成一个可连接的序列。同时该序列不会立刻发送事件，只有在调用 connect 之后才会开始
    private func publish() {
//        let netObservable = Observable<Any>.create { observer in
//            sleep(2) // 模拟网络延迟
//            print("开始网络请求")
//
//            observer.onNext("请求到的网络数据")
//            observer.onNext("请求到的本地数据")
//
//            observer.onCompleted()
//
//            return Disposables.create {
//                print("销毁回调了")
//            }
//        }.publish()
//
//        netObservable
//            .subscribe(onNext: { print("订阅-1：\($0)") })
//            .disposed(by: disposeBag)
//
//        // 有时候不止一次网络订阅，因为数据可能用在不同的地方
//        // 所以再订阅一次，会出现什么问题?
////        netObservable
////            .subscribe(onNext: { print("订阅-2：\($0)") })
////            .disposed(by: disposeBag)
//
//        _ = netObservable.connect()
        
        // 每隔1s发送1个事件
        let interval = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).publish()
        
        // 第一个订阅者（立刻开始订阅）
        _ = interval
            .subscribe(onNext: { print("订阅1：\($0)") })
        
        // 相当于把事件消息推迟了2s
        delay(2) {
            _ = interval.connect()
        }
        
        // 第二个订阅者，延迟5s开始订阅
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅2：\($0)") })
        }
    }
    
    /// replay
    /// 同上面的 publish 方法
    /// 相同之处在于：会将将一个正常的序列转换成一个可连接的序列。同时该序列不会立刻发送事件，只有在调用 connect 之后才会开始
    /// 不同在于：新的订阅者还能接收到订阅之前的事件消息（数量由设置的 bufferSize 决定）
    private func replay() {
        // 每隔1s发送1个事件
        let interval = Observable<Int>.interval(.seconds(1), scheduler: MainScheduler.instance).replay(5)
        
        _ = interval
            .subscribe(onNext: { print("订阅1: \($0)") })
        
        // 相当于把事件消息推迟了2s
        delay(2) { _ = interval.connect() }
        
        // 第二个订阅者，延迟5s开始订阅
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅2: \($0)") })
        }
    }
    
    /// multicast
    /// 将一个正常的序列转换成一个可连接的序列
    /// 还可以传入一个 Subject，每当序列发送事件时都会触发这个 Subject 的发送
    private func multicast() {
        // 创建一个Subject（后面的multicast()方法中传入）
        let subject = PublishSubject<Int>()
        
        _ = subject
            .subscribe(onNext: { print("Subject: \($0)") })
        
        // 每隔1s发送1个事件
        let interval = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).multicast(subject)
        
        // 第一个订阅者，立刻开始订阅
        _ = interval
            .subscribe(onNext: { print("订阅1: \($0)") })
        
        // 相当于把事件消息推迟了2s
        delay(2) {
            _ = interval.connect()
        }
        
        // 第二个订阅者，延迟5s开始订阅
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅2: \($0)") })
        }
    }
    
    /// refCount
    /// 将可被连接的 Observable 转换为普通 Observable
    /// 即该操作符可以自动连接和断开可连接的 Observable。当第一个观察者对可连接的Observable 订阅时，那么底层的 Observable 将被自动连接。
    /// 当最后一个观察者离开时，那么底层的 Observable 将被自动断开连接
    private func refCount() {
        let interval = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).publish().refCount()
        
        _ = interval
            .subscribe(onNext: { print("订阅1: \($0)") })
        
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅2: \($0)") })
        }
    }
    
    /// share
    /// 将使得观察者共享源 Observable，并且缓存最新的 n 个元素，将这些元素直接发送给新的观察者
    /// 简单来说 shareReplay 就是 replay 和 refCount 的组合
    private func share() {
        let interval = Observable<Int>.interval(RxTimeInterval.seconds(1), scheduler: MainScheduler.instance).share(replay: 5)
        
        _ = interval
            .subscribe(onNext: { print("订阅1: \($0)") })
        
        delay(5) {
            _ = interval
                .subscribe(onNext: { print("订阅2: \($0)") })
        }
    }
}
